package com.example.springbootkafka.service;

import com.example.springbootkafka.entity.User;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;

@Slf4j
@Service
public class KafkaProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final ObjectMapper objectMapper;

    @Autowired
    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate, ObjectMapper objectMapper) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = objectMapper;
    }

    public void sendMessage(String message) {
        log.info("KafkaProducer message:{}", message);
        //kafkaTemplate.send("test", message).addCallback();
        Future<SendResult<String, String>> future = kafkaTemplate.send("test", message);
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            try {
                future.get(); // 等待原始future完成
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

// 使用whenComplete方法
        completableFuture.whenComplete((result, ex) -> {
            if (ex != null) {
                System.out.println("Error occurred: " + ex.getMessage());
                // 成功发送
            } else {
                System.out.println("Completed successfully");
            }
        });

        /*future.whenComplete((result, ex) -> {
            if (ex == null) {
                // 成功发送
                RecordMetadata metadata = result.getRecordMetadata();
                System.out.println("Message sent successfully with offset: " + metadata.offset());
            } else {
                // 发送失败
                System.err.println("Failed to send message due to: " + ex.getMessage());
            }
        });*/


    }

    public void sendUser(User user) throws JsonProcessingException {
        //final ProducerRecord<String, String> record = createRecord(data);

        //ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", message);

        //ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", user);

        String userJson = objectMapper.writeValueAsString(user);
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test", userJson);
        /*future.addCallback(
                success -> System.out.println("Message sent successfully: " + userJson),
                failure -> System.err.println("Failed to send message: " + failure.getMessage())
        );*/
        CompletableFuture<Void> completableFuture = CompletableFuture.runAsync(() -> {
            try {
                future.get(); // 等待原始future完成
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        completableFuture.whenComplete((result, ex) -> {
            if (ex != null) {
                System.out.println("Error occurred: " + ex.getMessage());
                // 成功发送
            } else {
                System.out.println("Completed successfully");
            }
        });
    }
}